package com.my.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author zhupanlin
 * @version 1.0
 * @description: TODO
 * @date 2024/2/7 10:32
 */
public class FlowDemo {
    
    // 定义流中间操作处理器，只用写订阅者的接口
    static class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String>{
        private Flow.Subscription subscription;

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            System.out.println("processor订阅绑定完成");
            this.subscription = subscription;
            subscription.request(1);// 找上游要一个数据
            
        }

        @Override
        public void onNext(String item) {
            System.out.println("processor拿到数据：" + item);
            // 再加工
            item += "哈哈";
            // 把加工后的数据发出去
            submit(item);
            // 接收新数据
            subscription.request(1);

        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {
            System.out.println("完成咯");
        }
    }

    /**
     * 1. Publisher：发布者
     * 2. Subscriber：订阅者 
     * 3. Subscription：订阅关系
     * 4. Pro 
     * @param args
     */

    // 发布订阅模型：观察者模式
    public static void main(String[] args) {
        
        // 1.定义一个发布者：发布数据
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        
        // 2.定义一个中间操作
        MyProcessor myProcessor1 = new MyProcessor();
        MyProcessor myProcessor2 = new MyProcessor();
        
        //
        publisher.subscribe(myProcessor1);
        myProcessor1.subscribe(myProcessor2);
        
        
        // 3.定义一个订阅者：订阅者感兴趣发布者的数据
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            
            private Flow.Subscription subscription;
            
            // 在订阅时 onXXXX：在xxx事件发生时，执行这个回调
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread() + "订阅开始了：" +  subscription);
                this.subscription = subscription;
                
                // 从上游请求一个数据
                subscription.request(1);
            }

            // 在下一个元素到达时：执行这个回调； 接收到新数据
            @Override
            public void onNext(String item) {

                System.out.println(Thread.currentThread() + "订阅者，接收到新数据：" + item);
                subscription.request(1);

            }

            // 在错误发生时
            @Override
            public void onError(Throwable throwable) {

                System.out.println(Thread.currentThread() + "订阅者，接收到错误：" + throwable.getMessage());
                
            }

            @Override
            public void onComplete() {

                System.out.println(Thread.currentThread() + "订阅者，接收到完成信号：");
                
            }
        };
        
        // 将处理器和订阅者绑定
        myProcessor2.subscribe(subscriber);

        for (int i = 0; i < 10; i++) {
            // 发布10条数据
            publisher.submit("p-" + i);
            // publisher发布的所有数据在它的buffer区
        }

        try {
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        publisher.close();
        myProcessor1.close();
        myProcessor2.close();
        
       /* // 4.绑定发布者和订阅者
        publisher.subscribe(subscriber);

        for (int i = 0; i < 10; i++) {
            // 发布10条数据
            publisher.submit("p-" + i);
            // publisher发布的所有数据在它的buffer区
        }
        
        publisher.close();
        
        // 发布者有数据，订阅者就会拿到

        try {
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }*/


    }
    
}
